/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.good.akkaserver.server;

import java.util.concurrent.atomic.AtomicInteger;

import com.good.akkaserver.model.PlayerSession;
import com.good.akkaserver.msg.PlayerActorCreationMsg;
import com.good.akkaserver.msg.PlayerActorStopMsg;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

import static com.good.akkaserver.constant.AttributeMapConstant.NETTY_CHANNEL_KEY;  
/**
 * ConnectionEventHandler for Rpc.
 * 
 * @author jiangping
 * @version $Id: RpcConnectionEventHandler.java, v 0.1 2015-10-16 PM4:41:29 tao Exp $
 */
public class RpcConnectionEventHandler extends ConnectionEventHandler {
	
	private static AkkaServer akkaServer = AkkaServer.getInstance();
	private static ObjectPoolWorld objectPoolWorld = ObjectPoolWorld.getInstance();
	

	
	/** 用于生成  playerSessonID */
	private static final AtomicInteger PLAYER_SESSION = new AtomicInteger(1000);
	
    public RpcConnectionEventHandler() {
        super();
    }


    /**
     * @see com.alipay.remoting.ConnectionEventHandler#channelInactive(io.netty.channel.ChannelHandlerContext)
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        Attribute<PlayerSession> attr = ctx.attr(NETTY_CHANNEL_KEY);  
		PlayerSession session = attr.get();  
        if (session != null) {  
        	PlayerActorStopMsg msg = new PlayerActorStopMsg(session.getId());
         	akkaServer.stopPlayerActor(msg);
        }
     	
      
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
	    int id = PLAYER_SESSION.getAndIncrement();
	    Attribute<PlayerSession> attr = ctx.attr(NETTY_CHANNEL_KEY);  
	    PlayerSession session = new PlayerSession();
	    session.setId(id);
	    attr.setIfAbsent(session);  
		PlayerActorCreationMsg msg = new PlayerActorCreationMsg();
		msg.setSessionID(id);
		msg.setCtx(ctx);
		akkaServer.createPlayerActor(msg);
    }
}
